package com.fxu.framework.mq.redis.streams;

import cn.hutool.core.util.ByteUtil;
import com.fxu.framework.core.redis.RedisConfig;
import lombok.Data;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.Record;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.SerializationException;
import org.springframework.data.redis.stream.StreamListener;

import java.util.HashMap;
import java.util.Map;

import static com.fxu.framework.core.redis.RedisConfig.getJackson2JsonRedisSerializer;

/**
 * Redis Streams 消息队列
 * @author fangxu
 */
@Slf4j
public abstract class RedisStreamListener<K, V extends Record<K, ?>> implements StreamListener<K, V> {
    /**
     * 处理消息
     *
     * @param message 消息
     * @return 是否成功
     */
    public abstract boolean doMessage(Object message);

    /**
     * stream key
     */
    public abstract String getStreamKey();

    /**
     * 分组名
     */
    public abstract String getGroupName();

    /**
     * 消费者名
     */
    public abstract String getConsumerName();

    @Override
    public void onMessage(V message) {
        RecordId id = message.getId();
        K stream = message.getStream();
        System.out.printf("Received from stream %s, message ID: %s, body: %s%n", stream, id.getValue(), message.getValue());
        // 在这里处理消息并确认已消费
        try {
            //consumeMessageBefore(messageObj);
            // 消费消息
            boolean result = this.doMessage(message.getValue());
            // ack 消息消费完成
            if (result) {
                RedisStreamsHolder.getHolder().ack(getStreamKey(), getGroupName(), id.getValue());
            }
            // TODO 需要额外考虑以下几个点：
            // 1. 处理异常的情况
            // 2. 发送日志；以及事务的结合
            // 3. 消费日志；以及通用的幂等性
            // 4. 消费失败的重试，https://zhuanlan.zhihu.com/p/60501638
        } finally {
            //consumeMessageAfter(messageObj);
        }
    }
}
